package com.lyh.dt.server;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 全局事务处理
 * @author liuyuhong
 */
@Component
public class DtCoreService {

    @Autowired
    private DtActiveMqProducterService dtActiveMqProducterService;

    private Map<String, List<DtActiveMqBean>> transactionMap = new ConcurrentHashMap();

    private Map<String, Integer> transactionCount = new ConcurrentHashMap<>();

    private int cpuCoreNum = Runtime.getRuntime().availableProcessors();

    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(cpuCoreNum, cpuCoreNum * 2,
            0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));


    public Boolean transationHandle(DtActiveMqBean bean){
        if(bean.getCommandType() != null){
            switch (bean.getCommandType()){
                case 1:
                    List<DtActiveMqBean> transactionList = new ArrayList<>(16);
                    transactionMap.put(bean.getGroupId(), transactionList);
                    return true;
                case 2:
                    if(bean.getTransactionType() != null && transactionMap.containsKey(bean.getGroupId())){
                        List<DtActiveMqBean> dtActiveMqBeans = transactionMap.get(bean.getGroupId());
                        dtActiveMqBeans.add(bean);
                        //全部事务注册上来了时，进行处理
                        if(dtActiveMqBeans.size() == transactionCount.get(bean.getGroupId())){
                            threadPool.execute(()->{
                                transationCalculate(bean.getGroupId());
                            });
                        }
                        return true;
                    }
                    return false;
                case 3:
                    Integer integer = transactionCount.get(bean.getGroupId());
                    if(integer == null || integer == 0){
                        transactionCount.put(bean.getGroupId(), 1);
                    }else{
                        integer++;
                        transactionCount.put(bean.getGroupId(), integer);
                    }
                    return true;
            }
        }
        return false;
    }

    private void transationCalculate(String groupId){
        List<DtActiveMqBean> dtActiveMqBeans = transactionMap.get(groupId);
        List<DtActiveMqBean> rollbackTransation = dtActiveMqBeans.stream()
                .filter(dt -> dt.getTransactionType().equals(TransactionType.ROLLBACK)).collect(Collectors.toList());
        DtActiveMqBean bean;
        if(rollbackTransation.size() > 0){
            bean =  DtActiveMqBean.builder().commandType(4).groupId(groupId).transactionType(TransactionType.ROLLBACK).build();
        }else{
            bean =  DtActiveMqBean.builder().commandType(4).groupId(groupId).transactionType(TransactionType.COMMIT).build();
        }
        dtActiveMqProducterService.sendTopic(bean);
    }


}
